---
title: "Reference: Processor Interface | Processors"
description: "API reference for the Processor interface in Mastra, which defines the contract for transforming, validating, and controlling messages in agent pipelines."
---

# Processor Interface

The `Processor` interface defines the contract for all processors in Mastra. Processors can implement one or more methods to handle different stages of the agent execution pipeline.

## When processor methods run

The four processor methods run at different points in the agent execution lifecycle:

```
┌─────────────────────────────────────────────────────────────────┐
│                     Agent Execution Flow                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  User Input                                                     │
│      │                                                          │
│      ▼                                                          │
│  ┌─────────────────┐                                            │
│  │  processInput   │  ← Runs ONCE at start                      │
│  └────────┬────────┘                                            │
│           │                                                     │
│           ▼                                                     │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                   Agentic Loop                          │    │
│  │  ┌─────────────────────┐                                │    │
│  │  │  processInputStep   │  ← Runs at EACH step           │    │
│  │  └──────────┬──────────┘                                │    │
│  │             │                                           │    │
│  │             ▼                                           │    │
│  │       LLM Execution                                     │    │
│  │             │                                           │    │
│  │             ▼                                           │    │
│  │  ┌──────────────────────┐                               │    │
│  │  │ processOutputStream  │  ← Runs on EACH stream chunk  │    │
│  │  └──────────┬───────────┘                               │    │
│  │             │                                           │    │
│  │             ▼                                           │    │
│  │     Tool Execution (if needed)                          │    │
│  │             │                                           │    │
│  │             └──────── Loop back if tools called ────────│    │
│  └─────────────────────────────────────────────────────────┘    │
│           │                                                     │
│           ▼                                                     │
│  ┌─────────────────────┐                                        │
│  │ processOutputResult │  ← Runs ONCE after completion          │
│  └─────────────────────┘                                        │
│           │                                                     │
│           ▼                                                     │
│     Final Response                                              │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
```

| Method | When it runs | Use case |
|--------|--------------|----------|
| `processInput` | Once at the start, before the agentic loop | Validate/transform initial user input, add context |
| `processInputStep` | At each step of the agentic loop, before each LLM call | Transform messages between steps, handle tool results |
| `processOutputStream` | On each streaming chunk during LLM response | Filter/modify streaming content, detect patterns in real-time |
| `processOutputResult` | Once after generation completes | Post-process final response, log results |

## Interface definition

```typescript copy
interface Processor<TId extends string = string> {
  readonly id: TId;
  readonly name?: string;

  processInput?(args: ProcessInputArgs): Promise<ProcessInputResult> | ProcessInputResult;
  processOutputStream?(args: ProcessOutputStreamArgs): Promise<ChunkType | null | undefined>;
  processOutputResult?(args: ProcessOutputResultArgs): ProcessorMessageResult;
  processInputStep?(args: ProcessInputStepArgs): ProcessorMessageResult;
}
```

## Properties

<PropertiesTable
  content={[
    {
      name: "id",
      type: "string",
      description: "Unique identifier for the processor. Used for tracing and debugging.",
      isOptional: false,
    },
    {
      name: "name",
      type: "string",
      description: "Optional display name for the processor. Falls back to id if not provided.",
      isOptional: true,
    },
  ]}
/>

## Methods

### processInput

Processes input messages before they are sent to the LLM. Runs once at the start of agent execution.

```typescript copy
processInput?(args: ProcessInputArgs): Promise<ProcessInputResult> | ProcessInputResult;
```

#### ProcessInputArgs

<PropertiesTable
  content={[
    {
      name: "messages",
      type: "MastraDBMessage[]",
      description: "User and assistant messages to process (excludes system messages).",
      isOptional: false,
    },
    {
      name: "systemMessages",
      type: "CoreMessage[]",
      description: "All system messages (agent instructions, memory context, user-provided). Can be modified and returned.",
      isOptional: false,
    },
    {
      name: "messageList",
      type: "MessageList",
      description: "Full MessageList instance for advanced message management.",
      isOptional: false,
    },
    {
      name: "abort",
      type: "(reason?: string) => never",
      description: "Function to abort processing. Throws a TripWire error that stops execution.",
      isOptional: false,
    },
    {
      name: "tracingContext",
      type: "TracingContext",
      description: "Tracing context for observability.",
      isOptional: true,
    },
    {
      name: "requestContext",
      type: "RequestContext",
      description: "Request-scoped context with execution metadata like threadId and resourceId.",
      isOptional: true,
    },
  ]}
/>

#### ProcessInputResult

The method can return one of three types:

<PropertiesTable
  content={[
    {
      name: "MastraDBMessage[]",
      type: "array",
      description: "Transformed messages array. System messages remain unchanged.",
    },
    {
      name: "MessageList",
      type: "MessageList",
      description: "The same messageList instance passed in. Indicates you've mutated it directly.",
    },
    {
      name: "{ messages, systemMessages }",
      type: "object",
      description: "Object with both transformed messages and modified system messages.",
    },
  ]}
/>

---

### processInputStep

Processes input messages at each step of the agentic loop, before they are sent to the LLM. Unlike `processInput` which runs once at the start, this runs at every step including tool call continuations.

```typescript copy
processInputStep?(args: ProcessInputStepArgs): ProcessorMessageResult;
```

#### Execution order in the agentic loop

1. `processInput` (once at start)
2. `processInputStep` (at each step, before LLM call)
3. `prepareStep` callback (if provided)
4. LLM execution
5. Tool execution (if needed)
6. Repeat from step 2 if tools were called

#### ProcessInputStepArgs

<PropertiesTable
  content={[
    {
      name: "messages",
      type: "MastraDBMessage[]",
      description: "All messages including tool calls and results from previous steps.",
      isOptional: false,
    },
    {
      name: "messageList",
      type: "MessageList",
      description: "MessageList instance for managing messages.",
      isOptional: false,
    },
    {
      name: "stepNumber",
      type: "number",
      description: "Current step number (0-indexed). Step 0 is the initial LLM call.",
      isOptional: false,
    },
    {
      name: "systemMessages",
      type: "CoreMessage[]",
      description: "All system messages for read/modify access.",
      isOptional: false,
    },
    {
      name: "abort",
      type: "(reason?: string) => never",
      description: "Function to abort processing.",
      isOptional: false,
    },
    {
      name: "tracingContext",
      type: "TracingContext",
      description: "Tracing context for observability.",
      isOptional: true,
    },
    {
      name: "requestContext",
      type: "RequestContext",
      description: "Request-scoped context with execution metadata.",
      isOptional: true,
    },
  ]}
/>

#### Use cases

- Transforming message part types between providers (e.g., `reasoning` → `thinking` for Anthropic)
- Modifying messages based on step number or accumulated context
- Implementing per-step message filtering or enrichment
- Adding step-specific context or instructions

---

### processOutputStream

Processes streaming output chunks with built-in state management. Allows processors to accumulate chunks and make decisions based on larger context.

```typescript copy
processOutputStream?(args: ProcessOutputStreamArgs): Promise<ChunkType | null | undefined>;
```

#### ProcessOutputStreamArgs

<PropertiesTable
  content={[
    {
      name: "part",
      type: "ChunkType",
      description: "The current stream chunk being processed.",
      isOptional: false,
    },
    {
      name: "streamParts",
      type: "ChunkType[]",
      description: "All chunks seen so far in the stream.",
      isOptional: false,
    },
    {
      name: "state",
      type: "Record<string, unknown>",
      description: "Mutable state object that persists across chunks within a single stream.",
      isOptional: false,
    },
    {
      name: "abort",
      type: "(reason?: string) => never",
      description: "Function to abort the stream.",
      isOptional: false,
    },
    {
      name: "messageList",
      type: "MessageList",
      description: "MessageList instance for accessing conversation history.",
      isOptional: true,
    },
    {
      name: "tracingContext",
      type: "TracingContext",
      description: "Tracing context for observability.",
      isOptional: true,
    },
    {
      name: "requestContext",
      type: "RequestContext",
      description: "Request-scoped context with execution metadata.",
      isOptional: true,
    },
  ]}
/>

#### Return value

- Return the `ChunkType` to emit it (possibly modified)
- Return `null` or `undefined` to skip emitting the chunk

---

### processOutputResult

Processes the complete output result after streaming or generation is finished.

```typescript copy
processOutputResult?(args: ProcessOutputResultArgs): ProcessorMessageResult;
```

#### ProcessOutputResultArgs

<PropertiesTable
  content={[
    {
      name: "messages",
      type: "MastraDBMessage[]",
      description: "The generated response messages.",
      isOptional: false,
    },
    {
      name: "messageList",
      type: "MessageList",
      description: "MessageList instance for managing messages.",
      isOptional: false,
    },
    {
      name: "abort",
      type: "(reason?: string) => never",
      description: "Function to abort processing.",
      isOptional: false,
    },
    {
      name: "tracingContext",
      type: "TracingContext",
      description: "Tracing context for observability.",
      isOptional: true,
    },
    {
      name: "requestContext",
      type: "RequestContext",
      description: "Request-scoped context with execution metadata.",
      isOptional: true,
    },
  ]}
/>

## Processor types

Mastra provides type aliases to ensure processors implement the required methods:

```typescript copy
// Must implement processInput OR processInputStep (or both)
type InputProcessor = Processor & (
  | { processInput: required }
  | { processInputStep: required }
);

// Must implement processOutputStream OR processOutputResult (or both)
type OutputProcessor = Processor & (
  | { processOutputStream: required }
  | { processOutputResult: required }
);
```

## Usage examples

### Basic input processor

```typescript title="src/mastra/processors/lowercase.ts" showLineNumbers copy
import type { Processor, MastraDBMessage } from "@mastra/core";

export class LowercaseProcessor implements Processor {
  id = "lowercase";

  async processInput({ messages }): Promise<MastraDBMessage[]> {
    return messages.map((msg) => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map((part) =>
          part.type === "text"
            ? { ...part, text: part.text.toLowerCase() }
            : part
        ),
      },
    }));
  }
}
```

### Per-step processor with processInputStep

```typescript title="src/mastra/processors/reasoning-transformer.ts" showLineNumbers copy
import type { Processor, MastraDBMessage } from "@mastra/core";

export class ReasoningTransformer implements Processor {
  id = "reasoning-transformer";

  async processInputStep({ messages, messageList, stepNumber }) {
    // Transform reasoning parts to thinking parts at each step
    // This is useful when switching between model providers
    for (const msg of messages) {
      if (msg.role === "assistant" && msg.content.parts) {
        for (const part of msg.content.parts) {
          if (part.type === "reasoning") {
            (part as any).type = "thinking";
          }
        }
      }
    }
    return messageList;
  }
}
```

### Hybrid processor (input and output)

```typescript title="src/mastra/processors/content-filter.ts" showLineNumbers copy
import type { Processor, MastraDBMessage, ChunkType } from "@mastra/core";

export class ContentFilter implements Processor {
  id = "content-filter";
  private blockedWords: string[];

  constructor(blockedWords: string[]) {
    this.blockedWords = blockedWords;
  }

  async processInput({ messages, abort }): Promise<MastraDBMessage[]> {
    for (const msg of messages) {
      const text = msg.content.parts
        ?.filter((p) => p.type === "text")
        .map((p) => p.text)
        .join(" ");

      if (this.blockedWords.some((word) => text?.includes(word))) {
        abort("Blocked content detected in input");
      }
    }
    return messages;
  }

  async processOutputStream({ part, abort }): Promise<ChunkType | null> {
    if (part.type === "text-delta") {
      if (this.blockedWords.some((word) => part.textDelta.includes(word))) {
        abort("Blocked content detected in output");
      }
    }
    return part;
  }
}
```

### Stream accumulator with state

```typescript title="src/mastra/processors/word-counter.ts" showLineNumbers copy
import type { Processor, ChunkType } from "@mastra/core";

export class WordCounter implements Processor {
  id = "word-counter";

  async processOutputStream({ part, state }): Promise<ChunkType> {
    // Initialize state on first chunk
    if (!state.wordCount) {
      state.wordCount = 0;
    }

    // Count words in text chunks
    if (part.type === "text-delta") {
      const words = part.textDelta.split(/\s+/).filter(Boolean);
      state.wordCount += words.length;
    }

    // Log word count on finish
    if (part.type === "finish") {
      console.log(`Total words: ${state.wordCount}`);
    }

    return part;
  }
}
```

## Related

- [Processors overview](/docs/v1/agents/processors) - Conceptual guide to processors
- [Guardrails](/docs/v1/agents/guardrails) - Security and validation processors
- [Memory Processors](/docs/v1/memory/memory-processors) - Memory-specific processors
